package com.innodata.riskEngine.demo.producer;

import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.util.concurrent.ListenableFutureCallback;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RestController;

/**
 * @Author: liboshuai
 * @Date: 2023-07-28 10:22
 */
@Slf4j
@RestController
public class KafkaProducer {
    @Autowired
    private KafkaTemplate<String, Object> kafkaTemplate;

    /**
     * 发送消息到kafka
     *
     * @param message
     * @return
     */
    @GetMapping("/producer/normal/{message}")
    public String sendKafka(@PathVariable("message") String message) {
        kafkaTemplate.send("impSendOrder", message);
        log.info("发送message >>> {}", message);
        return "OK";
    }

    /**
     * 异步发送消息到kafka
     *
     * @param message
     * @return
     */
    @GetMapping("/producer/callbackOne/{message}")
    public String callbackOne(@PathVariable("message") String message) {
        kafkaTemplate.send("kafka-topic-test-callbackOne", message).addCallback(
                success -> {
                    String topic = success.getRecordMetadata().topic();
                    int partition = success.getRecordMetadata().partition();
                    long offset = success.getRecordMetadata().offset();
                    Object value = success.getProducerRecord().value();
                    log.info("消息被发送到 topic >>> {}, 分区 >>> {}, offset >>> {}, 消息内容为 >>> {};", topic, partition, offset, value);
                },
                failure -> log.info("消息发送失败 >>> {}; ", failure.getMessage()));
        return "Done!";
    }

    @GetMapping("/producer/callbackTwo/{message}")
    public String callbackTwo(@PathVariable("message") String message) {
        kafkaTemplate.send("kafka-topic-test-callbackTwo", message).addCallback(new ListenableFutureCallback<SendResult<String, Object>>() {
            @Override
            public void onFailure(Throwable ex) {
                log.info("消息发送失败 >>> {}; ", ex.getMessage());
            }

            @Override
            public void onSuccess(SendResult<String, Object> result) {
                String topic = result.getRecordMetadata().topic();
                int partition = result.getRecordMetadata().partition();
                long offset = result.getRecordMetadata().offset();
                Object value = result.getProducerRecord().value();
                log.info("消息被发送到 topic >>> {}, 分区 >>> {}, offset >>> {}, 消息内容为 >>> {};", topic, partition, offset, value);
            }
        });
        return "Done!";
    }

    @GetMapping("/producer/transactionOne/{message}")
    public String transactionOne(@PathVariable("message") String message) {
        // 声明事务，如果出现异常，则消息不会继续发送
        kafkaTemplate.executeInTransaction(operations -> {
            operations.send("kafka-topic-test-transactionOne", message);
            throw new RuntimeException("fail");
        });
        return "Done!";
    }

    @GetMapping("/producer/transactionTwo/{message}")
    public String transactionTwo(@PathVariable("message") String message) {
        // 声明事务，如果出现异常，消息仍会继续发送
        kafkaTemplate.send("kafka-topic-test-transactionTwo", message);
        throw new RuntimeException("fail");
    }
}
